package org.carl.base.driving.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.baomidou.lock.LockInfo;
import com.baomidou.lock.LockTemplate;
import org.carl.base.driving.annotation.EnableDriving;
import org.carl.base.driving.enums.ReceiveHandleStateEnum;
import org.carl.base.driving.enums.SendHandleStateEnum;
import org.carl.base.driving.DriveManage;
import org.carl.base.driving.bean.DrivingBeanConfig;
import org.carl.base.driving.listener.ReceiveListener;
import org.carl.base.driving.entity.ReceiveMessage;
import org.carl.base.driving.entity.SendMessage;
import org.carl.base.driving.core.ReceiveExecute;
import org.carl.base.driving.service.receive.ReceiveMessageService;
import org.carl.base.driving.core.SendExecute;
import org.carl.base.driving.service.send.SendMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * 消息驱动服务实现
 */
@Slf4j
@Component
@EnableScheduling
@ConditionalOnBean(value ={DrivingBeanConfig.class})
public class DriveManageImpl implements DriveManage {

    @Autowired
    private SendMessageService sendMessageService;

    @Autowired
    private ReceiveMessageService receiveMessageService;

    @Lazy
    @Autowired
    private SendExecute sendExecute;

    @Lazy
    @Autowired
    private ReceiveExecute receiveExecute;

    /**
     * 默认exchange
     */
    @Value("${spring.rabbitmq.template.exchange}")
    private String exchange;

    @Autowired
    private LockTemplate lockTemplate;

    @Autowired
    private DrivingBeanConfig drivingBeanConfig;

    /**
     * 单线程锁
     */
    private static final String MESSAGE_SERVER_LOCK = "MESSAGE:SERVER:LOCK:";

    @Bean
    public RabbitTemplate driveRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate driveRabbitTemplate = new RabbitTemplate(connectionFactory);
        driveRabbitTemplate.setExchange(exchange);
        driveRabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (correlationData == null || StringUtils.isEmpty(correlationData.getId())) {
                return;
            }
            if (ack) {
                // 修改消息发送成功
                sendMessageService.updateMessageState(correlationData.getId(), SendHandleStateEnum.SEND_SUCCESS.getCode());
                log.debug("drive消息确认 - 发送成功 ID{}", correlationData.getId());
            } else {
                sendMessageService.updateMessageState(correlationData.getId(), SendHandleStateEnum.SEND_FAIL.getCode());
                log.error("drive消消息确认 - 发送失败 ID{}", correlationData.getId());
            }
        });
        return driveRabbitTemplate;
    }

    @Override
    public void send(String eventName, String data) {
        sendExecute.execute(eventName, drivingBeanConfig.getRoutingKey(), data);
    }

    @Override
    public void send(String eventName, String routingKey, String data) {
        sendExecute.execute(eventName, routingKey, data);
    }

    @Override
    public void send(String eventName, String data, long sendTtl) {
        DateTime dateTime= DateUtil.offsetSecond(new Date(),(int) (sendTtl / 1000));
        sendExecute.expireExecute(eventName, drivingBeanConfig.getRoutingKey(), data, dateTime.toJdkDate());
    }


    @Override
    public void send(String eventName, String routingKey, String data, long sendTtl) {
        DateTime dateTime= DateUtil.offsetSecond(new Date(),(int) (sendTtl / 1000));
        sendExecute.expireExecute(eventName, routingKey, data, dateTime.toJdkDate());
    }

    @Override
    public void send(String eventName, String data, Date sendAt) {
        sendExecute.expireExecute(eventName, drivingBeanConfig.getRoutingKey(), data, sendAt);
    }

    @Override
    public void send(String eventName, String routingKey, String data, Date sendAt) {
        sendExecute.expireExecute(eventName, routingKey, data, sendAt);
    }

    @Bean
    public MessageListenerContainer transactionalListenerContainer(ConnectionFactory connectionFactory, ReceiveListener receiveListener) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(drivingBeanConfig.getMessageQueueName());
        container.setMessageListener(receiveListener);
        //初始化消费者数量
//        container.setConcurrentConsumers(32);
        //最大消费者数量
//        container.setMaxConcurrentConsumers(64);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return container;
    }

    /**
     * 未发送补发
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void synSend() {
        log.debug("消息驱动Send发送补偿，开始");
        String key = MESSAGE_SERVER_LOCK + "synSend";
        // 获取锁
        final LockInfo lockInfo = lockTemplate.lock(key, 3000L,5000L);
        if (null == lockInfo) {
            return;
        }
        try{
            DateTime dateTime = DateUtil.offsetSecond(new Date(), -30);
            // 查询一分钟之前发送状态为待确认的
            List<Integer> sendStateList = new ArrayList<>();
            sendStateList.add(SendHandleStateEnum.SEND_FAIL.getCode());
            sendStateList.add(SendHandleStateEnum.WAITING_CONFIRM.getCode());
            List<SendMessage> sendMessagesListSelect = sendMessageService.findOverdueOrder(sendStateList, dateTime.toDateStr());
            log.debug("消息驱动Send发送补偿查询结果:{}", sendMessagesListSelect);
            if (sendMessagesListSelect.isEmpty()) {
                return;
            }

            // 调用发送执行方法
            for (SendMessage sendMessage : sendMessagesListSelect) {
                SendMessage message = sendMessageService.selectById(sendMessage.getSendId());
                log.debug("消息驱动Send发送补偿:{}", message.getSendId());
                CompletableFuture.runAsync(() -> sendExecute.sendMessage(message));
//                driveThreadPoolExecutor.execute(() -> sendExecute.sendMessage(message));
            }
        } finally {
            //释放锁
            lockTemplate.releaseLock(lockInfo);
        }
    }

    /**
     * 同步接收记录
     */
    @Scheduled(cron = "0 0/1 * * * ?")
    public void synReceive() {
        log.debug("消息驱动Receive业务补偿，开始");
        String key = MESSAGE_SERVER_LOCK + "synReceive";
        // 获取锁
        final LockInfo lockInfo = lockTemplate.lock(key, 3000L,5000L);
        if (null == lockInfo) {
            return;
        }
        DateTime dateTime = DateUtil.offsetSecond(new Date(), -30);
        try{
            // 查询一分钟之前发送状态为初始
            List<Integer> receiveStateList = new ArrayList<>();
            receiveStateList.add(ReceiveHandleStateEnum.INIT.getCode());
            List<ReceiveMessage> receiveMessageList = receiveMessageService.findOverdueOrder(receiveStateList, dateTime.toDateStr());
            log.debug("消息驱动Receive业务补偿查询结果:{}", receiveMessageList);
            if (receiveMessageList.isEmpty()) {
                return;
            }

            // 调用接收执行方法
            for (ReceiveMessage receiveMessage : receiveMessageList) {
                log.debug("消息驱动Receive业务补偿:{}", receiveMessage.getReceiveId());
                // 异步执行业务，忽略执行结果
                CompletableFuture.runAsync(() -> receiveExecute.execute(receiveMessage.getReceiveId(), receiveMessage.getHandleName()));
//                driveThreadPoolExecutor.execute(() -> receiveExecute.execute(receiveMessage.getReceiveId(), receiveMessage.getHandleName()));
            }
        } finally {
            //释放锁
            lockTemplate.releaseLock(lockInfo);
        }
    }

    /**
     * 删除发送成功记录
     */
    @Scheduled(cron = "0 0/10 * * * ?")
    public void synDeleteSend() {
        String key = MESSAGE_SERVER_LOCK + "synDeleteSend";
        // 获取锁
        final LockInfo lockInfo = lockTemplate.lock(key, 3000L,5000L);
        if (null == lockInfo) {
            return;
        }
        try{
            DateTime dateTime = DateUtil.offsetHour(new Date(), -72);
            List<Integer> sendStateList = new ArrayList<>();
            sendStateList.add(SendHandleStateEnum.SEND_SUCCESS.getCode());
            List<SendMessage> sendMessagesListSelect = sendMessageService.findOverdueOrder(sendStateList, dateTime.toDateStr());
            if (sendMessagesListSelect.isEmpty()) {
                return;
            }

            // 调用发送执行方法
            for (SendMessage sendMessage : sendMessagesListSelect) {
                sendMessageService.deleteById(sendMessage.getSendId());
            }
        } finally {
            //释放锁
            lockTemplate.releaseLock(lockInfo);
        }
    }

    /**
     * 删除接收成功记录
     */
    @Scheduled(cron = "0 0/10 * * * ?")
    public void synDeleteReceive() {
        String key = MESSAGE_SERVER_LOCK + "synDeleteReceive";
        // 获取锁
        final LockInfo lockInfo = lockTemplate.lock(key, 3000L,5000L);
        if (null == lockInfo) {
            return;
        }
        try{
            DateTime dateTime = DateUtil.offsetHour(new Date(), -72);
            List<Integer> receiveStateList = new ArrayList<>();
            receiveStateList.add(ReceiveHandleStateEnum.PROCESS_SUCCESS.getCode());
            List<ReceiveMessage> receiveMessageList = receiveMessageService.findOverdueOrder(receiveStateList, dateTime.toDateStr());
            if (receiveMessageList.isEmpty()) {
                return;
            }
            // 调用接收执行方法
            for (ReceiveMessage receiveMessage : receiveMessageList) {
                receiveMessageService.deleteById(receiveMessage.getReceiveId());
            }
        } finally {
            //释放锁
            lockTemplate.releaseLock(lockInfo);
        }

    }
}
